
package com.trydofor.id.xnet;

import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;

import org.apache.mina.core.future.ConnectFuture;
import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.transport.socket.SocketConnector;
import org.apache.mina.transport.socket.nio.NioSocketConnector;
import org.slf4j.LoggerFactory;

import com.trydofor.aloe.exception.Exceptions;
import com.trydofor.id.byta.Bytable;
import com.trydofor.id.byta.Head;
import com.trydofor.id.byta.Packet;
import com.trydofor.id.core.A9Loader;
import com.trydofor.id.logs.Eventable;
import com.trydofor.id.logs.Eventer;
import com.trydofor.id.logs.LogLevel;
import com.trydofor.id.mina.LoggingFilter;
import com.trydofor.id.mina.PacketDecoder;
import com.trydofor.id.mina.PacketEncoder;

/**
 * @param <T> the Type
 * @author : Shi Rongjiu (www.trydofor.com)
 */
public abstract class NetLoader<T> implements A9Loader<List<T>>, Eventable {
    
    /**
     * the events
     */
    public enum Event implements Eventer.Event {
        /**
         * loader exception
         */
        LOADER_EXCEPTION,
        /**
         * the reply is NOT matched the fetching
         */
        MESSAGE_UNMATCHED,
        /**
         * session initial
         */
        SESSION_INITIAL,
        /**
         * session timeout
         */
        SESSION_TIMEOUT,
        /**
         * session idles too much
         */
        SESSION_IDLEMUCH
    }
    
    private final Eventer                    eventer;
    private final SocketConnector            connector;
    private final AtomicReference<IoSession> iosession;
    private final AtomicInteger              proccount;
    private final AtomicLong                 dealcount;
    private final String                     host;
    private final int                        port;
    private final int                        auth;
    
    private transient long                   timeout = Constant.TIMEOUT_MSECONDS;
    private transient int                    maxIdle = Constant.IDLEBOTH_SECONDS;
    
    /**
     * create a new connection to the server.
     * 
     * @param host the server host
     * @param port the server port
     * @param auth the server auth code
     * @throws NullPointerException if the host is null
     */
    public NetLoader(String host, int port, int auth) {
        Exceptions.checkNullPointer(host);
        //
        this.host = host;
        this.port = port;
        this.auth = auth;
        //
        eventer = new Eventer(LoggerFactory.getLogger(NetLoader.class));
        eventer.register(Event.LOADER_EXCEPTION, LogLevel.ERROR);
        eventer.register(Event.MESSAGE_UNMATCHED, LogLevel.WARN);
        eventer.register(Event.SESSION_IDLEMUCH, LogLevel.INFO);
        eventer.register(Event.SESSION_TIMEOUT, LogLevel.WARN);
        eventer.register(Event.SESSION_INITIAL, LogLevel.INFO);
        
        //
        PacketDecoder decoder = new PacketDecoder();
        decoder.getEventer().mergeEventAndRefer(eventer);
        registerDecoder(decoder);
        ProtocolCodecFilter codecFilter = new ProtocolCodecFilter(new PacketEncoder(), decoder);
        LoggingFilter logingFilter = new LoggingFilter(eventer);
        //
        connector = new NioSocketConnector();
        connector.getFilterChain().addLast("codec", codecFilter);
        connector.getFilterChain().addLast("logger", logingFilter);
        connector.setHandler(new InnerHandler());
        
        iosession = new AtomicReference<IoSession>();
        proccount = new AtomicInteger(0);
        dealcount = new AtomicLong(0);
    }
    
    /**
     * share the connection to the server.
     * 
     * @param loader the shared loader
     * @throws NullPointerException if the loader is null
     */
    public NetLoader(NetLoader<T> loader) {
        Exceptions.checkNullPointer(loader);
        //
        host = loader.host;
        port = loader.port;
        auth = loader.auth;
        eventer = loader.eventer;
        connector = loader.connector;
        iosession = loader.iosession;
        proccount = loader.proccount;
        dealcount = loader.dealcount;
        //
        timeout = loader.timeout;
        maxIdle = loader.maxIdle;
    }
    
    /**
     * {@inheritDoc}
     */
    public List<T> heavyLoad() throws Exception {
        
        final long seq = dealcount.incrementAndGet();
        final List<T> result = new ArrayList<T>();
        final Head head = new Head(auth, seq);
        
        final Packet packet = new Packet(head, makeRequestBody(head));
        final CountDownLatch waiter = new CountDownLatch(1);
        //
        IoSession session = iosession.get(); // atomic
        while (session == null || session.isClosing() || !session.isConnected()) {
            synchronized (connector) { // DDL works well here !?
                session = iosession.get(); // atomic
                if (session == null || session.isClosing() || !session.isConnected()) {
                    ConnectFuture connectFuture = connector.connect(new InetSocketAddress(host, port));
                    connectFuture.awaitUninterruptibly(timeout);
                    session = connectFuture.getSession(); // null
                    session.getConfig().setBothIdleTime(maxIdle);
                    eventer.log(Event.SESSION_INITIAL, "initial a new session={}", session);
                    iosession.set(session);
                }
            }
        }
        //
        final String rk = genResultKey(head);
        final String wk = genWaiterKey(head);
        try {
            proccount.incrementAndGet();
            session.setAttribute(rk, result);
            session.setAttribute(wk, waiter);
            session.write(packet);
            boolean done = waiter.await(timeout, TimeUnit.MILLISECONDS); // wait
            if (!done) {
                eventer.log(Event.SESSION_TIMEOUT, "waiting timeout={},session={},packet={}", timeout,
                        session, packet);
            }
        }
        finally {
            session.removeAttribute(rk);
            session.removeAttribute(wk);
            proccount.decrementAndGet();
        }
        
        return result;
    }
    
    /**
     * {@inheritDoc}
     */
    public void handleException(Exception e) {
        eventer.log(Event.LOADER_EXCEPTION, e, "loader exception\n");
    }
    
    /**
     * {@inheritDoc}
     */
    public Eventer getEventer() {
        return eventer;
    }
    
    /**
     * @return the server host
     */
    public String getHost() {
        return host;
    }
    
    /**
     * @return the server port
     */
    public int getPort() {
        return port;
    }
    
    /**
     * @return the server auth
     */
    public int getAuth() {
        return auth;
    }
    
    // ------------- setters and getters -------------
    /**
     * @return the timeout (millisecond)
     */
    public long getTimeout() {
        return timeout;
    }
    
    /**
     * @param timeout the timeout to set (millisecond)
     */
    public void setTimeout(long timeout) {
        this.timeout = timeout;
    }
    
    /**
     * @return the maxIdle (second)
     */
    public int getMaxIdle() {
        return maxIdle;
    }
    
    /**
     * @param maxIdle the maxIdle(second) to set
     */
    public void setMaxIdle(int maxIdle) {
        this.maxIdle = maxIdle;
    }
    
    // -----------------
    
    private String genResultKey(Head head) {
        return "Result@" + head.getDealCode();
    }
    
    private String genWaiterKey(Head head) {
        return "Waiter@" + head.getDealCode();
    }
    
    private class InnerHandler extends IoHandlerAdapter {
        
        @Override
        public void messageReceived(IoSession session, Object message) throws Exception {
            Packet reply = (Packet) message; // throw
            Head head = reply.getHead();
            try {
                @SuppressWarnings("unchecked")
                List<T> result = (List<T>) session.getAttribute(genResultKey(head));
                if (result == null) { // maybe timeout
                    eventer.log(Event.MESSAGE_UNMATCHED, "drop the unmatched,session={},head={}", session,
                            head);
                }
                else {
                    makeResultList(result, reply);
                }
            }
            finally {
                CountDownLatch waiter = (CountDownLatch) session.getAttribute(genWaiterKey(head));
                if (waiter != null) { // maybe timeout
                    waiter.countDown();
                }
            }
        }
        
        @Override
        public void exceptionCaught(IoSession session, Throwable cause) throws Exception {
            // let logging filter do
        }
        
        @Override
        public void sessionIdle(IoSession session, IdleStatus status) throws Exception {
            if (proccount.get() <= 0) {
                eventer.log(Event.SESSION_IDLEMUCH, "closing idle session={},idle={}", session,
                        session.getConfig().getBothIdleTime());
                session.close(true);
            }
        }
    }
    
    /* =========================== */

    /**
     * register some decoder, invoked once in the constructor.
     * 
     * @param pd the main decoder
     */
    protected abstract void registerDecoder(PacketDecoder pd);
    
    /**
     * make the request body, invoked once in every request.
     * 
     * @param head the packet head
     * @return the packet body
     */
    protected abstract Bytable<?> makeRequestBody(Head head);
    
    /**
     * make result by the reply, invoked once in every response.
     * 
     * @param result the result list
     * @param reply the reply packet
     */
    protected abstract void makeResultList(List<T> result, Packet reply);
}
